package com.gitee.sparta.examples.normal;

import cn.hutool.core.thread.GlobalThreadPool;
import com.alibaba.fastjson.JSON;
import com.gitee.sparta.annotation.FinalRetry;
import com.gitee.sparta.annotation.SpartaListener;
import com.gitee.sparta.api.SpartaTemplate;
import com.gitee.sparta.core.bean.SpartaChannel;
import com.gitee.sparta.examples.TopicConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author zyred
 * @since v 1.0.0
 */
@Slf4j
@RestController
@RequestMapping("/single")
public class SingleListener {
    private final Random random = new Random();

    private AtomicInteger a = new AtomicInteger(0);

    @SuppressWarnings("all")
    @Autowired private SpartaTemplate<Object> spartaTemplate;

    //@SpartaListener(topics = {TopicConstant.NORMAL_TOPIC})
    public void normalListener (SingleEntity singleEntity, SpartaChannel channel) {
        if (Objects.equals(singleEntity.getModel(), Model.normal)) {
            a.incrementAndGet();
            log.info("[normal consume] 延迟： {} ms entity: {}, count : {}",
                    (System.currentTimeMillis() - channel.getExpire()),
                    JSON.toJSONString(singleEntity), a.get());
            channel.ack();
        } else if (Objects.equals(singleEntity.getModel(), Model.retry)) {
            try {
                Integer a = null;
                a.intValue();
            } catch (Exception ex) {
                // 设置 5s 后重试, 重试次数超过了最大次数后，会触发 {@see SingleListener#normalFinalRetry} 的调用
                channel.setRetryIntervalTime(5_000);
                long nextMillis = System.currentTimeMillis() + channel.getRetryIntervalTime();
                Date date = new Date(nextMillis);
                log.info("[触发重试] 下次执行时间：{} ", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(date));
                throw ex;
            }
        } else if (Objects.equals(singleEntity.getModel(), Model.renew)) {
            channel.setRenewIntervalTime(40_000);
            long nextMillis = System.currentTimeMillis() + channel.getRetryIntervalTime();
            Date date = new Date(nextMillis);
            log.info("[触发续期] 下次执行时间：{} ", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(date));
            channel.renewal();
        }
    }

    @SpartaListener(topics = {TopicConstant.NORMAL_TOPIC})
    public void normalListener1 (String singleEntity, SpartaChannel channel) {
        System.out.println(singleEntity);
        channel.ack();
    }

    /**
     * 最终失败通知
     *
     * @param topic     topic
     * @param entity    消息体
     */
    @FinalRetry(topics = {TopicConstant.NORMAL_TOPIC})
    public void normalFinalRetry (String topic, SingleEntity entity) {
        log.info("[normal final retry] {}, topic : {}", JSON.toJSONString(entity), topic);
    }


    /**
     *  模拟插入消息
     *  http://localhost:9605/single/put?size=1&scope=1000&model=normal
     *
     * @param size      执行 n 次循环
     * @param scope     延迟随机数，生成随机的延迟时间
     * @param model     三种模式：正常消费，重试，续期
     */
    @GetMapping("/put")
    public void put (Long size, int scope, Model model) {
        for (int i = 0; i < size; i++) {
            int rand = random.nextInt(10000);

            SingleEntity normal = new SingleEntity();
            normal.setUserId(String.valueOf(rand));
            normal.setModel(model);

            this.spartaTemplate.writeAsync(TopicConstant.NORMAL_TOPIC, normal, random.nextInt(scope));
        }
    }

    /**
     *  压力测试
     *  http://localhost:9605/single/run
     */
    @GetMapping("/run")
    public void run () {
        GlobalThreadPool.execute(() -> {
            int count = 0;
            // 总共运行 5 分钟
            long end = System.currentTimeMillis() + 100_000;
            // 插入 500 条停顿 2 秒
            int a = 0, max = 10, standstill = 2;
            SingleEntity normal;
            for (;;) {

                int rand = random.nextInt(10000);
                if (rand == 0) {
                    rand  = 1000;
                }

                normal = new SingleEntity();
                normal.setUserId(String.valueOf(rand));
                normal.setModel(Model.normal);
                this.spartaTemplate.writeAsync(TopicConstant.NORMAL_TOPIC, normal, rand);
                count++;

                if (a == max) {
                    try {
                        TimeUnit.SECONDS.sleep(standstill);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    a = 0;
                }

                if (System.currentTimeMillis() >= end ) {
                    System.out.println("总写入消息：" + count);
                    break;
                }
                a ++;
            }
        });
    }
}
